package com.guqj.nio.netty.serialization.protobuf;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;

/**
 * @Title SubscribeServer
 * @Description netty protobuf
 * @date 2016年11月17日
 */
public class SubscribeServer {

	public void bind(int port) throws InterruptedException {
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup wokerGroup = new NioEventLoopGroup();

		try {
			ServerBootstrap b = new ServerBootstrap();
			b.group(bossGroup, wokerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024)
					.childHandler(new ServerChannelInitializer());
			Channel c = b.bind(port).sync().channel();
			System.out.println("服务已经启动,端口:" + port);
			c.closeFuture().sync();
		} finally {
			bossGroup.shutdownGracefully();
			wokerGroup.shutdownGracefully();
			System.out.println("服务销毁!");
		}
	}

	class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
		@Override
		protected void initChannel(SocketChannel ch) throws Exception {
			ChannelPipeline p = ch.pipeline();
			// 处理半包
			p.addLast(new ProtobufVarint32FrameDecoder());
			p.addLast(new ProtobufDecoder(SubscribeReqProto.SubscribeReq.getDefaultInstance()));
			p.addLast(new ProtobufVarint32LengthFieldPrepender());
			p.addLast(new ProtobufEncoder());
			p.addLast(new SubscribeServerHandler());
		}
	}

	public static void main(String[] args) throws InterruptedException {
		new SubscribeServer().bind(11013);
	}
}
